package com.cskaoyan.order.mq.consumer;

import com.cskaoyan.order.dal.entitys.OrderItem;
import com.cskaoyan.order.dal.entitys.Stock;
import com.cskaoyan.order.dal.persistence.OrderItemMapper;
import com.cskaoyan.order.dal.persistence.OrderMapper;
import com.cskaoyan.order.dal.persistence.StockMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.util.List;

/**
 * @author gatsby
 * @since 2022/10/29 16:36
 */
@Slf4j
@Component
public class DelayOrderCancelConsumer {
    @Autowired
    OrderMapper orderMapper;

    @Autowired
    OrderItemMapper orderItemMapper;

    @Autowired
    StockMapper stockMapper;

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer("delay_order_cancel_consumer");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //接收到消息后判断订单状态
                log.info("准备接收延迟消息");
                MessageExt message = msgs.get(0);
                byte[] body = message.getBody();
                log.info("接收延迟消息 {}" ,body);
                try {
                    String orderId = new String(body, 0, body.length, "utf-8");
                    int status = orderMapper.selectStatusByOrderId(orderId);
                    log.info("订单的状态为 {}" ,status);
                    //1.如果是成功支付或者已取消，则什么都不用管
                    if (status == 4 || status == 7) {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    //2.如果订单状态是初始化状态，那么执行订单取消
                    if (status == 0) {
                        //  a.将订单状态改为已取消
                        log.info("执行订单取消的业务逻辑");
                        orderMapper.updateStatusByOrderId(orderId);
                        //  b.查询出订单对应的所有商品条目，根据商品条目修改对应的库存信息， -lock_count +stock_count
                        List<OrderItem> orderItems = orderItemMapper.queryByOrderId(orderId);
                        for (OrderItem orderItem : orderItems) {
                            Stock stock = new Stock();
                            stock.setLockCount(-orderItem.getNum());
                            stock.setStockCount(orderItem.getNum().longValue());
                            stockMapper.updateStock(stock);
                        }

                    }
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }


                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        try {
            //订阅主题
            consumer.subscribe("order_delay_cancel", "*");
            //启动consumer
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }

}
